package com.sam.kafka.interceptor;

import org.apache.kafka.clients.consumer.ConsumerInterceptor;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;

import java.util.Map;

/**
 * @author: wm.xue
 * @date: 2021/11/10 10:21
 * @description:
 */
public class MyConsumerInterceptorTwo implements ConsumerInterceptor<String,String> {
    @Override
    public ConsumerRecords<String, String> onConsume(ConsumerRecords<String, String> records) {
        // poll 方法返回结果之前最后要调用方法
        System.err.println("Two ----- Start");

        // 消息不做处理直接返回
        return records;
    }

    @Override
    public void onCommit(Map<TopicPartition, OffsetAndMetadata> offsets) {
        // 消费者提交偏移量时经过该方法
        System.err.println("Two ---- End");
    }

    @Override
    public void close() {
        // 关闭拦截器用到的资源操作，如打开的文件，数据库等
        // to do nothing
    }

    @Override
    public void configure(Map<String, ?> configs) {
    }
}
